package cn.z2huo.demo.kafka.spring.producer.producer;

import cn.z2huo.core.util.IdUtils;
import cn.z2huo.demo.kafka.spring.producer.jackson.KafkaJsonMapper;
import cn.z2huo.demo.kafka.spring.producer.properties.TestTopicProperties;
import cn.z2huo.demo.model.dataobject.user.UserDO;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.json.JsonMapper;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.RandomStringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;

/**
 * <p>
 *
 * @author z2huo
 */
@RequiredArgsConstructor
@Component
@Slf4j
public class TestTopicProducer {

    private final TestTopicProperties testTopicProperties;

    @Autowired
    @Qualifier("testTopicKafkaTemplate")
    private KafkaTemplate<String, String> kafkaTemplate;

    @Autowired
    @Qualifier("testTopicCustomConfigKafkaTemplate")
    private KafkaTemplate<String, String> customKafkaTemplate;

    public void batchSendUser(List<UserDO> userDOS) {
        JsonMapper jsonMapper = KafkaJsonMapper.INSTANCE;
        String topic = testTopicProperties.getTopicName();

        long start, end;

        start = System.currentTimeMillis();

        long totalBytes = 0L;
        try {
            for (int i = 0; i < userDOS.size(); i++) {
                log.info("send the NO.{} user", i);
                String userDOJsonString = jsonMapper.writeValueAsString(userDOS.get(i));
                int bytesLength = userDOJsonString.getBytes(StandardCharsets.UTF_8).length;
                totalBytes += bytesLength;
                log.info("userDO json String length is {}, size is {} bytes, userDOJsonString is {}", userDOJsonString.length(),
                        bytesLength, userDOJsonString);
                kafkaTemplate.send(topic, userDOJsonString);
            }
        } catch (JsonProcessingException e) {
            throw new RuntimeException(e);
        }
        end = System.currentTimeMillis();
        log.info("send {} users cost {} ms, all json string bytes size is {}", userDOS.size(), end - start, totalBytes);
    }

    /**
     * <p>测试一下 spring kafka 的生产者的自定义配置
     * 顺带测试一下 通过 CompletableFuture 接收来自 send 的结果
     *
     * @see cn.z2huo.demo.kafka.spring.config.producer.TestTopicProducerConfig#testTopicCustomConfigKafkaTemplate(ProducerFactory)
     *
     */
    public void batchSendUserCustomBatchSizeLingerConfig(List<UserDO> userDOS) {
        JsonMapper jsonMapper = KafkaJsonMapper.INSTANCE;
        String topic = testTopicProperties.getTopicName();

        long start, end;
        start = System.currentTimeMillis();
        long totalBytes = 0L;
        try {
            for (int i = 0; i < userDOS.size(); i++) {
                log.info("send the NO.{} user", i);
                String userDOJsonString = jsonMapper.writeValueAsString(userDOS.get(i));
                int bytesLength = userDOJsonString.getBytes(StandardCharsets.UTF_8).length;
                totalBytes += bytesLength;
                log.info("userDO json String length is {}, size is {} bytes, userDOJsonString is {}",
                        userDOJsonString.length(), bytesLength, userDOJsonString);
                CompletableFuture<SendResult<String, String>> completableFuture = customKafkaTemplate.send(topic, userDOJsonString);
                completableFuture.whenComplete((sendResult, throwable) -> {
                    if (throwable != null) {
                        log.error("send user error", throwable);
                    } else {
                        log.info("send user success, sendResult is {}", sendResult);
                    }
                });
            }
        } catch (JsonProcessingException e) {
            throw new RuntimeException(e);
        }
        end = System.currentTimeMillis();
        log.info("send {} users cost {} ms, all json string bytes size is {}", userDOS.size(), end - start, totalBytes);
    }

    public List<UserDO> generateUser(int count) {
        ArrayList<UserDO> userDOS = new ArrayList<>();
        for (int i = 0; i < count; i++) {
            UserDO userDO = new UserDO();
            userDO.setUserCode("userCode " + RandomStringUtils.randomAlphabetic(8))
                    .setUserName("userName " + RandomStringUtils.randomAlphabetic(8))
                    .setId(IdUtils.nextId())
                    .setCreateTime(LocalDateTime.now())
                    .setOperateTime(LocalDateTime.now())
                    .setValidDate(LocalDate.now())
                    .setInvalidDate(LocalDate.now().plusYears(1))
                    .setValidFlag("1")
                    .setDeleteFlag("0");
            userDOS.add(userDO);
        }
        log.info("generate user list success, user count is {}", count);
        return userDOS;
    }

}
